package com.shujia.hbase.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class Demo8Index4 {

    private HConnection connection;
    private HBaseAdmin hBaseAdmin;

    @Before
    public void init() {

        //创建hadoop配置文件对象
        Configuration configuration = new Configuration();

        //指定zk地址
        configuration.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");


        try {
            //建立连接   和zk的连接
            connection = HConnectionManager.createConnection(configuration);

            hBaseAdmin = new HBaseAdmin(configuration);

            System.out.println("连接建立成功");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void createTable() {
        HTableDescriptor student = new HTableDescriptor("student");
        HColumnDescriptor info = new HColumnDescriptor("info");
        //info.setMaxVersions(10);
        //info.setTimeToLive(100);

        //增加列簇
        student.addFamily(info);

        try {
            //创建表
            hBaseAdmin.createTable(student);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Test
    public void dropTable() {

        //删除表
        try {
            if (hBaseAdmin.tableExists("student")) {
                hBaseAdmin.disableTable("student");
                hBaseAdmin.deleteTable("student");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void putAll() {

        //批量插入数据
        ArrayList<Put> puts = new ArrayList<Put>();

        try {

            //获取表的实例
            HTableInterface student = connection.getTable("student");


            BufferedReader bufferedReader = new BufferedReader(new FileReader("data/students.txt"));

            String line;
            while ((line = bufferedReader.readLine()) != null) {

                String[] split = line.split(",");

                String id = split[0];
                String name = split[1];
                Integer age = Integer.parseInt(split[2]);
                String gender = split[3];
                String clazz = split[4];

                //以班级和学号一起作为rowkey
                String rewkey = clazz + "_" + id;


                //构建put对象   一学号作为rowkey
                Put put = new Put(rewkey.getBytes());

                //增加列
                put.add("info".getBytes(), "name".getBytes(), name.getBytes("utf-8"));

                //Bytes.toBytes(age)  将基本数据类型转换成字节数组
                put.add("info".getBytes(), "age".getBytes(), Bytes.toBytes(age));

                put.add("info".getBytes(), "gender".getBytes(), gender.getBytes());
                //put.add("info".getBytes(), "clazz".getBytes(), clazz.getBytes());

                //将put放到集合中
                puts.add(put);

            }


            //批量插入数据
            student.put(puts);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Test
    public void scanClazz() {

        //获取所有性别为男的学生

        try {
            HTableInterface student = connection.getTable("student");
            Scan scan = new Scan();


            //前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("文科一班".getBytes());

            //创建过滤器   列值过滤器   需要进行全表扫描
            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

            //增加过滤器
            scan.setFilter(rowFilter);

            ResultScanner scanner = student.getScanner(scan);

            Result result;
            while ((result = scanner.next()) != null) {
                String sex = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));

                //在客户端进行过滤   还是会进行全表扫描
                //if ("男".equals(sex)) {
                print(result);
                //}
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 在由索引表的情况下查询数据
     */

    @Test
    public void queryIndex() {

        String clazz = "文科一班";

        /**
         * 1、先查询索引表
         * 2、再查询数据表
         *
         */


        try {
            HTableInterface student_index = connection.getTable("student_index");

            Get get = new Get(clazz.getBytes());
            Result result = student_index.get(get);

            ArrayList<Get> gets = new ArrayList<>();

            for (Cell cell : result.listCells()) {

                //取出列名  列名就是学号
                String id = Bytes.toString(CellUtil.cloneQualifier(cell));

                gets.add(new Get(id.getBytes()));

            }

            //查询学生表

            HTableInterface student = connection.getTable("student");

            //取出所有学生
            Result[] results = student.get(gets);


            for (Result result1 : results) {
                //取出rowkey
                String id = Bytes.toString(result1.getRow());


                // 1、直接通过列名取值   如果列名不存在 返回对应类型的默认值
                byte[] bytes = result1.getValue("info".getBytes(), "name".getBytes());
                //通过String的构造函数 构建字符串
                //String name = new String(bytes, "utf-8");

                //hbase  推荐方式
                String name = Bytes.toString(bytes);

                //int  类型  Bytes.toInt  转换
                int age = Bytes.toInt(result1.getValue("info".getBytes(), "age".getBytes()));

                String gender = Bytes.toString(result1.getValue("info".getBytes(), "gender".getBytes()));
                String clazz1 = Bytes.toString(result1.getValue("info".getBytes(), "clazz".getBytes()));

                System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz1);

            }


        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    public static void print(Result result) {
        //取出rowkey
        String id = Bytes.toString(result.getRow());

        String[] split = id.split("_");


        // 1、直接通过列名取值   如果列名不存在 返回对应类型的默认值
        byte[] bytes = result.getValue("info".getBytes(), "name".getBytes());
        //通过String的构造函数 构建字符串
        //String name = new String(bytes, "utf-8");

        //hbase  推荐方式
        String name = Bytes.toString(bytes);

        //int  类型  Bytes.toInt  转换
        int age = Bytes.toInt(result.getValue("info".getBytes(), "age".getBytes()));
        String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));

        System.out.println(split[1] + "\t" + name + "\t" + age + "\t" + gender + "\t" + split[0]);
    }

}
